package org.example.seckilljava.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.example.seckilljava.common.ResponseEnum;
import org.example.seckilljava.common.SKStatusEnum;
import org.example.seckilljava.dto.SecKillDTO;
import org.example.seckilljava.dto.SeckillMessage;
import org.example.seckilljava.entity.*;
import org.example.seckilljava.mapper.*;
import org.example.seckilljava.service.KafkaProducerService;
import org.example.seckilljava.service.SecKillService;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class SecKillServiceImpl implements SecKillService {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private ObjectMapper objectMapper;
    @Resource
    GoodsMapper goodsMapper;
    @Resource
    UserQuotaMapper userQuotaMapper;
    @Resource
    QuotaMapper quotaMapper;
    @Resource
    SeckillStockMapper seckillStockMapper;
    @Resource
    OrderMapper orderMapper;
    @Resource
    SeckillRecordMapper seckillRecordMapper;
    @Resource
    private KafkaProducerService kafkaProducerService;

    // Redis key前缀
    private static final String GOODS_STOCK_KEY = "seckill:goods:stock:";
    private static final String USER_QUOTA_KEY = "seckill:user:quota:";
    private static final String GLOBAL_QUOTA_KEY = "seckill:global:quota:";
    private static final String GOODS_INFO_KEY = "seckill:goods:info:";

    private DefaultRedisScript<Long> seckillScript;

    @PostConstruct
    public void init() {
        seckillScript = new DefaultRedisScript<>();
        seckillScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/seckill.lua")));
        seckillScript.setResultType(Long.class);
    }

    /**
     * 秒杀服务 v1
     * @param secKillDTO 秒杀参数
     * @return
     */
    @Override
    @Transactional
    public String doSecKill_V1(SecKillDTO secKillDTO) {
        String goodsNum = secKillDTO.getGoodsNum();
        Integer num = secKillDTO.getNum();
        Long userID = secKillDTO.getUserId();
        // 1. 拿到商品信息
        Goods goods = goodsMapper.getGoodsByNum(goodsNum);
        Long goodsID = goods.getId();
        // 2. 判断是否满足用户限额和商品限额
        Boolean quotaFlag = judgeQuota_v1(userID, goodsID, num);
        if (!quotaFlag) { // 回滚
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            return ResponseEnum.ERR_GOODS_STOCK_NOT_ENOUGH.message();
        }
        // 3. 操作数据库，完成秒杀
        // 3.1. 操作用户限额表，使其killed_num增长，乐观锁，解决并发状态的问题
        Integer affectedRows = userQuotaMapper.incrKilledNum(userID, goodsID, num);
        if (affectedRows == 0) {
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            return ResponseEnum.ERR_GOODS_STOCK_NOT_ENOUGH.message();
        }
        // 3.2. 操作全局限额表，使其killed_num增长
        affectedRows = quotaMapper.incrKilledNum(goodsID, num);
        if (affectedRows == 0) {
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            return ResponseEnum.ERR_GOODS_STOCK_NOT_ENOUGH.message();
        }
        // 3.4. 在 t_seckill_stock 库存表中减去秒杀商品的数量（sql语句实现一个乐观锁，当库存大于等于num时才操作）
        affectedRows = seckillStockMapper.descStock(goodsID, num);
        if (affectedRows == 0) { // 回滚
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            return ResponseEnum.ERR_GOODS_STOCK_NOT_ENOUGH.message();
        }
        // 4. 生成订单：生成订单编号和秒杀记录编号（UUID）
        String secNum = UUID.randomUUID().toString();
        String orderNum = UUID.randomUUID().toString();
        Order order = saveOrder(userID, goods.getSeller(), goodsID, goods.getGoodsNum(), num, goods.getPrice(), orderNum, SKStatusEnum.SK_STATUS_BEFORE_PAY.getValue());
        // 5. 生成秒杀记录
        saveSeckillRecord(userID, goodsID, order.getPrice(), orderNum, secNum, SKStatusEnum.SK_STATUS_BEFORE_PAY.getValue());
        // 6. 返回订单号
        return orderNum;
    }

    /**
     * 秒杀服务 v2
     * @param secKillDTO 秒杀参数
     * @return
     */
    @Override
    public String doSecKill_V2(SecKillDTO secKillDTO) {
        String goodsNum = secKillDTO.getGoodsNum();
        Integer num = secKillDTO.getNum();
        Long userID = secKillDTO.getUserId();

        // 1. 从Redis中获取商品信息
        Goods goods = getGoodsFromRedis(goodsNum);
        if (goods == null) {
            goods = goodsMapper.getGoodsByNum(goodsNum);
            updateGoodsInRedis(goods);
        }
        Long goodsID = goods.getId();

        // 2. 使用Lua脚本进行原子性操作，在redis中判别库存
        // 执行lua脚本
        List<String> keys = Arrays.asList(goodsID.toString(), userID.toString());
        Long result = redisTemplate.execute(seckillScript, keys, num);
        if (result == null || result < 0) {
            String errorMsg;
            switch (result != null ? result.intValue() : -1) {
                case -1:
                    errorMsg = "商品库存信息不存在";
                    break;
                case -2:
                    errorMsg = "商品库存不足";
                    break;
                case -3:
                    errorMsg = "用户限额不足";
                    break;
                case -4:
                    errorMsg = "全局限额不足";
                    break;
                default:
                    errorMsg = "未知错误";
            }
            log.error("秒杀失败: {}", errorMsg);
            return ResponseEnum.ERR_GOODS_STOCK_NOT_ENOUGH.message();
        }

        // 3. 生成订单编号和秒杀记录编号
        String secNum = UUID.randomUUID().toString();
        String orderNum = UUID.randomUUID().toString();

        // 4. 创建订单对象
        Order order = new Order(goods.getSeller(), userID, goodsID, goods.getGoodsNum(), num, orderNum, goods.getPrice() * num, SKStatusEnum.SK_STATUS_BEFORE_PAY.getValue());

        // 5. 创建秒杀记录对象
        SeckillRecord seckillRecord = new SeckillRecord(userID, goodsID, secNum, orderNum, order.getPrice(), SKStatusEnum.SK_STATUS_BEFORE_PAY.getValue());

        // 6. 创建消息对象
        SeckillMessage message = new SeckillMessage(orderNum, secNum, order, seckillRecord, System.currentTimeMillis());

        // 7. 发送消息到Kafka
        log.info("正在发送秒杀消息: {}", message);
        kafkaProducerService.sendSeckillMessage(message);

        return orderNum;
    }

    /**
     * 是否满足限额（用户限额或商品限额）
     * @param userID
     * @param goodsID
     * @param num
     * @return
     */
    private Boolean judgeQuota_v1(long userID, long goodsID, Integer num) {
        Integer userKilledNum; // 用户已经秒杀的商品数
        Integer userQuotaNum; // 用户秒杀限额数
        Integer globalKilledNum; // 全局已经秒杀的商品数
        Integer globalQuotaNum; // 全局商品限额数
        int userFQ = -1, globalFQ = -1;
        // 1. 拿到用户限额数据和商品限额数据
        UserQuota userQuota = userQuotaMapper.getUserGoodsUserQuota(userID, goodsID);
        Quota globalQuota = quotaMapper.getGoodsQuota(goodsID);
        // 2. 如果有数据，复制
        if (userQuota == null) {
            userQuota = new UserQuota();
            userQuota.setGoodsId(goodsID);
            userQuota.setKilledNum(0);
            userQuota.setNum(1); // 默认限额一个
            userQuota.setUserId(userID);
            userQuotaMapper.save(userQuota);
        }
        if (globalQuota == null) {
            globalQuota = new Quota();
            globalQuota.setGoodsId(goodsID);
            globalQuota.setNum(-1); // 默认不限额
            globalQuota.setKilledNum(0);
            quotaMapper.save(globalQuota);
        }
        userKilledNum = userQuota.getKilledNum();
        userQuotaNum = userQuota.getNum();
        globalKilledNum = globalQuota.getKilledNum();
        globalQuotaNum = globalQuota.getNum();
        // 3. 确定限额情况
        if (userQuotaNum != -1) {
            userFQ = userQuotaNum - userKilledNum;
        }
        if (globalQuotaNum != -1) {
            globalFQ = globalQuotaNum - globalKilledNum;
        }
        // 4. 所有情况
        if (userFQ == -1 && globalFQ == -1)
            return true;
        if (globalFQ == -1) // 如果 global 不限额
            return userFQ >= num;
        if (userFQ == -1) // 如果 user 不限额
            return globalFQ >= num;
        return userFQ >= num && globalFQ >= num;
    }

    /**
     * 保存订单
     * @param userID
     * @param seller
     * @param goodsID
     * @param goodsNum
     * @param num
     * @param price
     * @param orderNum
     * @param status
     * @return
     */
    private Order saveOrder(Long userID, Long seller, Long goodsID, String goodsNum, Integer num, Float price, String orderNum, Integer status) {
        Order order = new Order(seller, userID, goodsID, goodsNum, num, orderNum, price * num, status);
        orderMapper.save(order);
        return order;
    }

    /**
     * 保存秒杀记录
     * @param userID
     * @param goodsID
     * @param price
     * @param orderNum
     * @param secNum
     * @param status
     */
    private void saveSeckillRecord(Long userID, Long goodsID, Float price, String orderNum, String secNum, Integer status) {
        SeckillRecord record = new SeckillRecord(userID, goodsID, secNum, orderNum, price, status);
        seckillRecordMapper.save(record);
    }

    /**
     * 从Redis中获取商品库存
     */
    private Integer getGoodsStockFromRedis(Long goodsId) {
        String key = GOODS_STOCK_KEY + goodsId;
        Object value = redisTemplate.opsForValue().get(key);
        return value != null ? (Integer) value : null;
    }

    /**
     * 更新Redis中的商品库存
     */
    private void updateGoodsStockInRedis(Long goodsId, Integer stock) {
        String key = GOODS_STOCK_KEY + goodsId;
        redisTemplate.opsForValue().set(key, stock, 24, TimeUnit.HOURS);
    }

    /**
     * 从Redis中获取用户限额
     */
    private Integer getUserQuotaFromRedis(Long userId, Long goodsId) {
        String key = USER_QUOTA_KEY + userId + ":" + goodsId;
        Object value = redisTemplate.opsForValue().get(key);
        return value != null ? (Integer) value : null;
    }

    /**
     * 更新Redis中的用户限额
     */
    private void updateUserQuotaInRedis(Long userId, Long goodsId, Integer quota) {
        String key = USER_QUOTA_KEY + userId + ":" + goodsId;
        redisTemplate.opsForValue().set(key, quota, 24, TimeUnit.HOURS);
    }

    /**
     * 从Redis中获取全局限额
     */
    private Integer getGlobalQuotaFromRedis(Long goodsId) {
        String key = GLOBAL_QUOTA_KEY + goodsId;
        Object value = redisTemplate.opsForValue().get(key);
        return value != null ? (Integer) value : null;
    }

    /**
     * 更新Redis中的全局限额
     */
    private void updateGlobalQuotaInRedis(Long goodsId, Integer quota) {
        String key = GLOBAL_QUOTA_KEY + goodsId;
        redisTemplate.opsForValue().set(key, quota, 24, TimeUnit.HOURS);
    }

    /**
     * 从Redis中获取商品信息
     */
    private Goods getGoodsFromRedis(String goodsNum) {
        String key = GOODS_INFO_KEY + goodsNum;
        Object value = redisTemplate.opsForValue().get(key);
        if (value == null) return null;

        // 转成 JSON，再转成 Goods
        Goods goods = objectMapper.convertValue(value, Goods.class);
        return goods;
    }

    /**
     * 更新Redis中的商品信息
     */
    private void updateGoodsInRedis(Goods goods) {
        String key = GOODS_INFO_KEY + goods.getGoodsNum();
        redisTemplate.opsForValue().set(key, goods, 24, TimeUnit.HOURS);
    }
} 